package com.honeybee.cloud.framework.rocket.consumer;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.MQPullConsumer;
import org.apache.rocketmq.client.consumer.MQPullConsumerScheduleService;
import org.apache.rocketmq.client.consumer.PullResult;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.util.Assert;

import java.util.List;

/**
 * @Description 一些mq拉取基础的消费者
 * @Author DELL
 * @Create 2019-04-03
 * @Since 1.0.0
 */
@Slf4j
public abstract class AbstractPullConsumer implements InitializingBean {

    /**
     * 消费者组
     */
    protected String consumerGroup = "DEFAULT_CONSUMER_GROUP";

    /**
     * mq连接路径
     */
    protected String nameSrvAddress;

    /**
     * tags
     */
    protected String tags = "*";

    /**
     * topic
     */
    protected String topic;

    /**
     * maxNums
     */
    protected Integer maxNums = 16;
    /**
     * delayTimeMillis
     */
    protected Integer delayTimeMillis = 1000;

    /**
     * threadNums
     */
    protected Integer threadNums = 5;
    /**
     * fromStore
     */
    protected Boolean fromStore = false;
    /**
     * messageModel
     */
    protected MessageModel messageModel = MessageModel.CLUSTERING;

    /**
     * 初始化
     *
     * @throws Exception
     */
    @Override
    public abstract void afterPropertiesSet() throws Exception;


    /**
     * 使用时请调用此方法实例化，完成相关配置
     *
     * @param nameSrvAddress
     * @param topic
     * @param tags
     * @param consumerGroup
     */
    protected void init(String nameSrvAddress, String topic, String tags, String consumerGroup) {
        init(nameSrvAddress, topic, tags, consumerGroup, null, null, null, null, null);
    }

    /**
     * @param nameSrvAddress
     * @param topic
     * @param tags
     * @param consumerGroup
     * @param delayTimeMillis
     * @param maxNums
     * @param threadNums
     * @param messageModel
     * @param fromStore
     */
    protected void init(String nameSrvAddress, String topic, String tags, String consumerGroup, Integer delayTimeMillis, Integer maxNums, Integer threadNums, MessageModel messageModel, Boolean fromStore) {
        this.nameSrvAddress = nameSrvAddress;
        this.topic = topic;
        this.tags = tags;
        this.consumerGroup = consumerGroup;
        this.delayTimeMillis = delayTimeMillis;
        this.maxNums = maxNums;
        this.threadNums = threadNums;
        this.messageModel = messageModel;
        this.fromStore = fromStore;
    }

    /**
     * 处理事件
     *
     * @param messageExtList
     */
    protected abstract void event(List<MessageExt> messageExtList);

    /**
     * 开启消费者
     *
     * @throws MQClientException
     */
    protected void startConsumerGroup() throws MQClientException {
        log.info("开始启动消费组:{},地址:{}", consumerGroup, nameSrvAddress);
        Assert.hasLength(nameSrvAddress, "nameSrvAddress is not null");
        Assert.hasLength(topic, "topic is not null");
        Assert.hasLength(tags, "tags is not null");
        Assert.notNull(delayTimeMillis, "delayTimeMillis is not null");
        Assert.notNull(maxNums, "maxNums is not null");
        //消费组
        final MQPullConsumerScheduleService scheduleService = new MQPullConsumerScheduleService(topic);
        //MQ NameService地址
        scheduleService.getDefaultMQPullConsumer().setNamesrvAddr(nameSrvAddress);
        //负载均衡模式
        scheduleService.setMessageModel(messageModel);
        scheduleService.setPullThreadNums(threadNums);
        //需要处理的消息topic
        scheduleService.registerPullTaskCallback(topic, (mq, context) -> {
            MQPullConsumer consumer = context.getPullConsumer();
            try {
                long offset = consumer.fetchConsumeOffset(mq, fromStore);
                if (offset < 0) {
                    offset = 0;
                }
                //一次拉一条
                PullResult pullResult = consumer.pull(mq, tags, offset, maxNums);
                switch (pullResult.getPullStatus()) {
                    case FOUND: {
                        List<MessageExt> messageExtList = pullResult.getMsgFoundList();
                        event(messageExtList);
                        consumer.updateConsumeOffset(mq, pullResult.getNextBeginOffset());
                        break;
                    }
                    default: {
                        log.debug("其他状态:{}", pullResult.getPullStatus());
                        break;
                    }
                }
            } catch (Exception e) {
                log.error("mq消费异常", e);
            } finally {
                context.setPullNextDelayTimeMillis(delayTimeMillis);
            }
        });
        scheduleService.start();
        log.info("成功启动消费组:{}", consumerGroup);
    }


}
